package com.example.kafka.config;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.TimeWindows;

import java.time.Duration;

/**
 *  Kafka Streams 实现 日志实时聚合
 *  每分钟统计一次不同日志级别（如 INFO、ERROR、WARN）的日志数量，并输出到 Kafka 的新主题 log-level-stats，同时打印控制台信息。
 * @author wkj
 */
//@Configuration
// EnableKafkaStreams:启用 Spring Kafka Streams 功能
//@EnableKafkaStreams
public class KafkaStreamConfig {

    /**
     * 定义一个 Kafka 流处理逻辑的 Bean，处理来自 Kafka 的日志流（KStream）。
     * StreamsBuilder 是 Kafka Streams 构建处理拓扑的类。
     */
//    @Bean
    public KStream<String, String> kStream(StreamsBuilder builder) {
        // 从名为 application-logs 的 Kafka topic 中读取数据（格式为 JSON 字符串）
        KStream<String, String> stream = builder.stream("application-logs");

        ObjectMapper objectMapper = new ObjectMapper();

        // 提取 level 字段并统计
        stream
                .map((key, value) -> {
                    try {
                        JsonNode node = objectMapper.readTree(value);
                        // 提取 "level" 字段
                        String level = node.get("level").asText();
                        // 返回键为 level，值为 1
                        return new KeyValue<>(level, "1");
                    } catch (Exception e) {
                        // 解析失败则标记为 UNKNOWN
                        return new KeyValue<>("UNKNOWN", "1");
                    }
                })
                // 分组 + 滑动窗口聚合统计
                // .groupByKey()：按日志级别（INFO、ERROR 等）分组。
                .groupByKey()
                // .windowedBy(...)：基于时间窗口（这里是 1 分钟）统计。
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
                // .count()：对每个日志级别在每个 1 分钟内出现的次数进行计数。
                .count()
                .toStream()
                // 输出控制台调试信息：每分钟内每种日志级别的数量。
                .peek((key, count) -> System.out.printf("窗口[%s] 等级[%s] 数量: %d%n",
                        key.window().startTime(),
                        key.key(),
                        count))
                // 写回 Kafka：发送到 log-level-stats 主题
                .map((key, count) -> new KeyValue<>(key.key(), count.toString()))
                // 写入到 Kafka 的 log-level-stats topic 中，供下游消费或存入 Elasticsearch。
                .to("log-level-stats", Produced.with(Serdes.String(), Serdes.String()));

        // 返回源数据流（不是聚合流），通常用于后续处理或测试，不影响核心逻辑。
        return stream;
    }
}
